package com.demo.debezium.listener;

import com.alibaba.fastjson.JSON;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.Executors;

@Slf4j
@Component
public class MysqlListener {

    private final List<DebeziumEngine<ChangeEvent<String, String>>> engineList = new ArrayList<>();


    /**
     * 通过配置创建监听
     */
    private MysqlListener(@Qualifier("mysqlProperties") List<Properties> list) {
        for (Properties props : list) {
            this.engineList.add(DebeziumEngine.create(Json.class)
                    .using(props)
                    .notifying(record -> {
                        receive(record.value(), props.getProperty("name"));
                    }).build());
        }
    }

    private void receive(String value, String name) {
        if (Objects.nonNull(value)) {
            Map<String, Object> valueMap = JSON.parseObject(value, Map.class);
            Map<String, Object> payload = JSON.parseObject(JSON.toJSONString(valueMap.get("payload")), Map.class);

            String after = String.valueOf(payload.get("after"));
            String before = String.valueOf(payload.get("before"));
            String op = String.valueOf(payload.get("op"));
            log.info("数据库名称：{}",name);
            log.info("修改前数据：{}",before);
            log.info("修改后数据：{}",after);
            log.info("操作：{}",op);


        }
    }

    /**
     * 项目启动，创建Debezium引擎
     */
    @PostConstruct
    private void start() {
        for (DebeziumEngine<ChangeEvent<String, String>> engine : engineList) {
            Executors.newSingleThreadExecutor().execute(engine);
        }
    }

    /**
     * 项目销毁时关闭Debezium引擎
     */
    @PreDestroy
    private void stop() {
        for (DebeziumEngine<ChangeEvent<String, String>> engine : engineList) {
            if (engine != null) {
                try {
                    engine.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

}
